package com.jacky.ws.redis;

import com.corundumstudio.socketio.handler.AuthorizeHandler;
import com.corundumstudio.socketio.namespace.Namespace;
import com.corundumstudio.socketio.namespace.NamespacesHub;
import com.corundumstudio.socketio.protocol.JsonSupport;
import com.corundumstudio.socketio.store.RedissonStore;
import com.corundumstudio.socketio.store.RedissonStoreFactory;
import com.corundumstudio.socketio.store.Store;
import com.corundumstudio.socketio.store.pubsub.*;
import org.redisson.api.RedissonClient;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.Set;
import java.util.UUID;

public class CustomRedissonStoreFactory extends RedissonStoreFactory {

    private final Logger LOG = LoggerFactory.getLogger(CustomRedissonStoreFactory.class);

    private final RedissonClient redissonClient;
    private final RedissonClient redisPub;
    private final RedissonClient redisSub;
    private final PubSubStore pubSubStore;

    public CustomRedissonStoreFactory(RedissonClient redisson) {
        // 这里一定要调用一下父类的构造方法,把redissonClient传过去,
        // 不然父类的默认构造方法会创建一个127.0.0.1:6379的连接,如果不是连接本地的redis就会报错
        super(redisson);
        this.redissonClient = redisson;
        this.redisPub = redisson;
        this.redisSub = redisson;
        this.pubSubStore = new CustomRedissonPubSubStore(redisPub, redisSub, this.getNodeId());
    }

    @Override
    public Store createStore(UUID sessionId) {
        return new RedissonStore(sessionId, this.redissonClient);
    }

    public PubSubStore pubSubStore() {
        return this.pubSubStore;
    }

    @Override
    public void shutdown() {
        this.redissonClient.shutdown();
        this.redisPub.shutdown();
        this.redisSub.shutdown();
    }

    @Override
    public void init(NamespacesHub namespacesHub, AuthorizeHandler authorizeHandler, JsonSupport jsonSupport) {
        this.pubSubStore().subscribe(PubSubType.DISCONNECT, msg -> LOG.debug("{} sessionId:{}", PubSubType.DISCONNECT, msg.getNodeId()), DisconnectMessage.class);

        this.pubSubStore().subscribe(PubSubType.CONNECT, msg -> {
            authorizeHandler.connect(msg.getSessionId());
            LOG.debug("{} sessionId:{}", PubSubType.CONNECT, msg.getSessionId());
        }, ConnectMessage.class);

        this.pubSubStore().subscribe(PubSubType.DISPATCH, msg -> {
            // todo 这里是消息分发的逻辑,可根据具体业务做相应修改
            String name = msg.getRoom();
            Namespace ns = namespacesHub.get(msg.getNamespace());
            if (ns != null) {
                LOG.info("dispatch,room:{},nameSpace:{}", name, ns.getName());
                ns.dispatch(name, msg.getPacket());
            }
            LOG.debug("{} packet:{}", PubSubType.DISPATCH, msg.getPacket());
        }, DispatchMessage.class);

        this.pubSubStore().subscribe(PubSubType.JOIN, msg -> {
            String name = msg.getRoom();
            Namespace ns = namespacesHub.get(msg.getNamespace());
            if (ns != null) {
                ns.join(name, msg.getSessionId());
            }
            LOG.debug("{} sessionId:{}", PubSubType.JOIN, msg.getSessionId());
        }, JoinLeaveMessage.class);

        this.pubSubStore().subscribe(PubSubType.BULK_JOIN, msg -> {
            Set<String> rooms = msg.getRooms();
            for (String room : rooms) {
                Namespace ns = namespacesHub.get(msg.getNamespace());
                if (ns != null) {
                    ns.join(room, msg.getSessionId());
                }
            }
            LOG.debug("{} sessionId:{}", PubSubType.BULK_JOIN, msg.getSessionId());
        }, BulkJoinLeaveMessage.class);

        this.pubSubStore().subscribe(PubSubType.LEAVE, new PubSubListener<JoinLeaveMessage>() {
            @Override
            public void onMessage(JoinLeaveMessage msg) {
                String room = msg.getRoom();
                Namespace ns = namespacesHub.get(msg.getNamespace());
                if (ns != null) {
                    ns.leave(room, msg.getSessionId());
                }
                LOG.debug("{} sessionId:{}", PubSubType.LEAVE, msg.getSessionId());
            }
        }, JoinLeaveMessage.class);

        this.pubSubStore().subscribe(PubSubType.BULK_LEAVE, msg -> {
            Set<String> rooms = msg.getRooms();
            for (String room : rooms) {
                Namespace ns = namespacesHub.get(msg.getNamespace());
                if (ns != null) {
                    ns.leave(room, msg.getSessionId());
                }
            }
            LOG.debug("{} sessionId:{}", PubSubType.BULK_LEAVE, msg.getSessionId());
        }, BulkJoinLeaveMessage.class);
    }
}
